package com.test.cn.kafka

import akka.actor.{Actor, ActorSystem, Props}
import com.test.cn.kafka.message._
import com.typesafe.config.ConfigFactory

import scala.collection.mutable
import scala.concurrent.duration.FiniteDuration

class KafkaServerMaster(host: String, port: Int) extends Actor {

  val workerSet = new mutable.HashSet[WorkerInfo]()

  val workerMap = new mutable.HashMap[String, WorkerInfo]()


  //启动之前的方法
  override def preStart(): Unit = {

    import scala.concurrent.duration._
    import context.dispatcher
    //启动定时检查
    context.system.scheduler.schedule(5000 millis, 5000 millis, self, CheckTimeOutWork)

  }

  override def receive: Receive = {
    case RegisterWork(id: String, memory: Int, cores: Int) => {
      val info = new WorkerInfo(id, memory, cores)
      workerSet += info
      workerMap += (id -> info)
      sender() ! RegisteredWorker(host + ":" + port)
    }
    case HeartBeat(id) => {
      val currentTime = System.currentTimeMillis()
      val workerInfo = workerMap(id)
      workerInfo.lastHeartBeatTime = currentTime
      workerMap(id) = workerInfo
    }
    case CheckTimeOutWork => {
      val cur = System.currentTimeMillis
      //过滤超时的client
      val deadWorker = workerSet.filter(x=>cur-x.lastHeartBeatTime>15000)
      //从记录删除删除超时的worker
      for (w <- deadWorker){
        workerMap -= w.id
        workerSet -= w
      }
      println("当前注册成功的节点数" + workerSet.size)

    }


  }


}

object KafkaServerMaster extends App {

  val host = "127.0.0.1"
  val port = 9999

  //构造配置参数值，使用3个双引号可以多行，使用s可以在字符串中使用类似Shell的变量
  val str =
    s"""
       |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
       |akka.remote.netty.tcp.hostname =${host}
       |akka.remote.netty.tcp.port=${port}
      """.stripMargin

  println(str)
  //通过工厂方法获得一个config对象
  val conf = ConfigFactory.parseString(str)
  //初始化一个ActorSystem，其名为master
  val actorSystem = ActorSystem("masterSystem", conf)
  //使用actorSystem实例化一个名为Master的actor,注意这个名称在Worker连接Master时会用到
  actorSystem.actorOf(Props(new KafkaServerMaster(host, port)), "masterActor")


}